package com.atguigu.flink.wordcount;

import com.atguigu.flink.pojo.WordCount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/3/31
 *
 *
 *  非法类型异常
 *InvalidTypesException:
 *      调用flatMap后，由于类型擦除，无法决定返回值的类型。
 *      The return type of function 'main(Demo7_LamdaPOJODemo.java:32)' could not be determined automatically, due to type erasure.
 *       可以在转换调用的结尾，调用returns() 告知 类型信息
 *      You can give type information hints by using the returns(...) method on the result of the transformation call,
 *      或需要让函数实现 ResultTypeQueryable，不采取。
 *      or by letting your function implement the 'ResultTypeQueryable' interface
 *
 */
public class Demo7_LamdaPOJODemo
{
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(1);

        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888);

        SingleOutputStreamOperator<WordCount> ds2 = ds
            .flatMap((FlatMapFunction<String, WordCount>) (line, out) -> {

                String[] words = line.split(" ");
                for (String word : words) {
                    out.collect(new WordCount(word, 1));
                }
            })
            //使用returns明确告知 Collector<T> 中返回的类型是 WordCount
            .returns(WordCount.class);

        ds2
            .keyBy((KeySelector<WordCount, String>) value -> value.getWord())
            //流中的数据类型是POJO，此时聚合时，写对应的属性名
            .sum("count")
            .print();


        env.execute();

    }
}
